package com.zhang.hadoop.flink.test3;

import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @author: zhang yufei
 * @createTime:2022/5/15 11:49
 * @description:
 */
public class TransformFilterTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<Event> events = new ArrayList<>();
        events.add(new Event("yanghui", "./yindao", 1000L));
        events.add(new Event("yuping", "./siwajiao", 2000L));
        events.add(new Event("yangdan", "./gangmen", 2000L));
        events.add(new Event("jingru", "./niaodao", 2000L));
        DataStreamSource<Event> stream = env.fromCollection(events);

        //1.传入一个实现了FilterFunction的类的对象
        SingleOutputStreamOperator<Event> result1 = stream.filter(new MyFilter());
        result1.print();

        //2.传入一个匿名类实现FilterFunction接口
        SingleOutputStreamOperator<Event> result2 = stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event event) throws Exception {
                return event.user.endsWith("yuping");
            }
        });
        result2.print();

        //3.传入lambda表达式
        stream.filter(data -> data.user.endsWith("yangdan")).print("lambda: yangdan");

        env.execute();
    }

    public static class MyFilter implements FilterFunction<Event> {

        @Override
        public boolean filter(Event event) throws Exception {
            return event.user.endsWith("yanghui");
        }
    }
}
